//
// Created by Yucheng on 5/23/17.
//

#ifndef DIYLANE_WORKER_POOL_H
#define DIYLANE_WORKER_POOL_H

#include <vector>
#include <boost/thread.hpp>
#include "utilities.hpp"

namespace diylane {
    namespace pool {

        struct Worker {
            bool busy = false;
            uint64_t timestamp = 0;

            Worker() {
                timestamp = util::S::utcMilliseconds();
            }

            virtual ~Worker() {

            }

            virtual bool isAvailable() {
                if (busy) {
                    return false;
                }
                return true;
            }

            void setBusy(bool v) {
                if (v) {
                    timestamp = util::S::utcMilliseconds();
                }
                busy = v;
            }
        };

        struct WorkerPool {
            int max_workers = 5;

        protected:
            std::string name = "Undefined";

        private:
            boost::mutex mutex;
            std::vector<Worker*> workers;
            bool pendingCreateOperation = false;
            boost::thread *workerThread = NULL;
            bool shutdownWorkerThread = false;

        public:
            WorkerPool() {

            }

            virtual ~WorkerPool() {
                if (workerThread) {
                    shutdownWorkerThread = true;    // fix leaks: 2017-07-03
                    workerThread->join();           //
                    delete workerThread;
                    workerThread = NULL;
                }
                // fix leaks: 2017-07-03
                for (size_t i=0; i<workers.size(); i++) {
                    Worker* w = workers[i];
                    if (w) {
                        delete w;
                        workers[i] = NULL;
                    }
                }
            }

            void release(Worker* w) {
                boost::lock_guard<boost::mutex> lock(mutex);
                if (w) {
                    w->setBusy(false);
                }
            }

        protected:
            /**
             * Create a real worker.
             * this method must be implemented.
             * @return Worker*
             */
            virtual Worker* createOne_(std::string& error) {
                return NULL;
            }

            /**
             * this is used to check bad or expired workers,
             * the method will be called onece per $? millisecond.
             */
            virtual void checkOne_() {

            }

            void setPendingCreateOperation(bool v) {
                boost::lock_guard<boost::mutex> lock(mutex);
                pendingCreateOperation = v;
            }

            bool hasPendingCreateOperation() {
                boost::lock_guard<boost::mutex> lock(mutex);
                return pendingCreateOperation;
            }

            void add_(Worker* w) {
                boost::lock_guard<boost::mutex> lock(mutex);
                workers.push_back(w);

                std::cout << "pool::WorkerPool<" << name << ">, added a new worker." << std::endl;

                // this is a good place to create a worker thread
                if (workerThread == NULL) {
                    workerThread = new boost::thread(&WorkerPool::runWorkerThread, this);
                }
            }

            size_t numberOfWorkers() {
                boost::lock_guard<boost::mutex> lock(mutex);
                return workers.size();
            }

            Worker* getAvailableOne() {
                boost::lock_guard<boost::mutex> lock(mutex);
                for (size_t i = 0; i < workers.size(); i++) {
                    Worker* w = workers[i];
                    if (w->isAvailable()) {
                        w->setBusy(true);
                        return w;
                    }
                }
                return NULL;
            }
            
            bool shouldCreateOne() {
                boost::lock_guard<boost::mutex> lock(mutex);
                if (workers.size() < max_workers && !pendingCreateOperation) {
                    pendingCreateOperation = true;
                    return true;
                }
                return false;
            }

            /**
             * Used for real worker pool.
             * @return <Worker*>
             */
            Worker* acquireWorker_(std::string& error) {
                Worker* w = NULL;

                // Find one is available
                while (true) {
                    w = getAvailableOne();
                    if (w) {
                        return w;
                    }

                    // create one if there is no pending creation
                    // and the number of connections is less than the max
                    if (shouldCreateOne()) {
                        Worker* created = createOne_(error);
                        if (created == NULL) {
                            setPendingCreateOperation(false);	// release creation mutex whatever
                            return NULL;
                        }
                        add_(created);
                        setPendingCreateOperation(false);
                    }

                    // yield for other tasks
                    boost::this_thread::sleep(boost::posix_time::milliseconds(1));
                }

                return NULL;
            }

            void runWorkerThread() {
                while (!shutdownWorkerThread) {
                    checkOne_();
                    boost::this_thread::sleep(boost::posix_time::milliseconds(500));
                }
            }
        };

        struct AcquiredWorker {
            WorkerPool* pool;
            Worker* w;

            AcquiredWorker() {
                pool = NULL;
                w = NULL;
            }

            virtual ~AcquiredWorker() {
                if (pool) {
                    pool->release(w);
                    pool = NULL;
                }
            }
        };
    }
}

#endif //DIYLANE_WORKER_POOL_H
